package streaming.api.tableapi;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * 从kafka读取数据  (SQL-DDL)
 *  输出到MySQL JDBC    (SQL-DDL)
 */
public class TableTest7_jdbc2 {

    public static void main(String[] args) throws Exception {
        // 1. 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        String topicFrom = "myTest";
        String kafkaServer = "192.168.36.130:9092,192.168.36.129:9092,192.168.36.128:9092";
        String kafkaFormat = "json";
        String kafkaConsumerGroupId = "flink-consumer-TableTest7_jdbc2";

        String db_url = "jdbc:mysql://192.168.36.129:3306/zzb?useUnicode=true&characterEncoding=UTF-8&useSSL=false";
        String db_driver = "com.mysql.cj.jdbc.Driver";
        String db_username = "root";
        String db_password = "root";
        String db_table = "flink_tg_test";

        StringBuffer sb1 = new StringBuffer();
        sb1.append("CREATE TABLE inputTable (");
        sb1.append("t_id STRING, t_key STRING, t_val STRING");
        sb1.append(") WITH (");
        sb1.append("'connector' = 'kafka', ");
        sb1.append("'topic' = '").append(topicFrom).append("', ");
        sb1.append("'properties.bootstrap.servers' = '").append(kafkaServer).append("', ");
        sb1.append("'properties.group.id' = '").append(kafkaConsumerGroupId).append("', ");
        sb1.append("'format' = '").append(kafkaFormat).append("'");
        // test
        sb1.append(")");
        System.out.println("inputTable SQL: " + sb1.toString());
        tableEnv.executeSql(sb1.toString());

        StringBuffer sb2 = new StringBuffer();
        sb2.append("CREATE TABLE outputTable (");
        sb2.append("tg_id VARCHAR(32),tg_key VARCHAR(100),tg_value VARCHAR(100) ");
        sb2.append(") WITH (");
        sb2.append("'connector' = 'jdbc', ");
        sb2.append("'url' = '").append(db_url).append("', ");
        sb2.append("'table-name' = '").append(db_table).append("', ");
        sb2.append("'driver' = '").append(db_driver).append("', ");
        sb2.append("'username' = '").append(db_username).append("', ");
        sb2.append("'password' = '").append(db_password).append("', ");
        sb2.append("'sink.buffer-flush.max-rows'   = '1'");
        sb2.append(")");
        System.out.println("outputTable SQL: " + sb2.toString());
        tableEnv.executeSql(sb2.toString());

        // 方式二 自定义SQl
        tableEnv.executeSql("INSERT INTO outputTable(tg_id, tg_key, tg_value) select t_id, t_key, t_val from inputTable");
    }

}
